package tresp.coordinator;

import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

import tresp.bthreads.BThreadType;
import tresp.medusa.Medusa;
import tresp.utils.BThreadManager;
import tresp.utils.BThreadVote;
import tresp.utils.ITankInfo;
import tresp.utils.TankInfoImpl;
import tresp.utils.logging.MyLogger;


/**
 * 
 * <b>AbstractLevelCoordinator</b>
 * 
 * <p>
 * 	This is a base class for all coordinator levels.
 * </p>
 * 
 * @author <font color="blue">Maxim Kirilov</font>
 *
 */
public class AbstractLevelCoordinator implements ICoordinator<BThreadVote> {

    private static Logger logger = MyLogger.getLogger( AbstractLevelCoordinator.class );
    
    /**Reference to the tank.*/
    protected Medusa medusa;
    
    /**Coordinator heirarchy level*/
    protected int level;
    
    /**indicates whether this coordinator is active or not*/
    protected boolean isActive = true;
    
    /**Holds this level b-threads votes*/
    private AbstractQueue<BThreadVote> lVotes;
    
    
    private Set<BThreadVote> blockedVotes;
    
    /**synchronizton lock.*/
    private Lock synchronizationLock;
    
    
    public AbstractLevelCoordinator(int level, Medusa medusa){
	this.level = level;
	synchronizationLock = new ReentrantLock( true );
	lVotes = new PriorityQueue<BThreadVote>();
	blockedVotes = Collections.synchronizedSet( new HashSet<BThreadVote>() );
	this.medusa = medusa;
	logger.info("Coordinator Level "+ getLevel() +" was started");
    }
    
    
    @Override
    public void run() {
	
	while( isActive ){
	    synchronizationLock.lock();
	    BThreadVote bThreadVote = getNextVote();
	    synchronizationLock.unlock();
	    
	    if ( bThreadVote != null ) { executeAction( bThreadVote ); }
	    try{
		Thread.sleep((3 - getLevel()) * 100L);
	    }catch (InterruptedException e) { logger.info("Coordinator Level "+ getLevel() +" was interrupted"); }
	}
	
    }

    @Override
    public BThreadVote getNextVote() {
	
	BThreadVote nextVote = getNextAllowedVote();
	if ( nextVote == null ){ return null;}
	deleteRedundantVotes(lVotes, nextVote);
	return nextVote;
    }


    private void deleteRedundantVotes(Collection<BThreadVote> lv, BThreadVote nextVote) {
	
	List<BThreadVote> lReduced = new LinkedList<BThreadVote>();
	for (BThreadVote vote : lv){
	    if (vote.getVoteType().equals( nextVote.getVoteType() )){
		lReduced.add( vote );
	    }
	}
	lv.removeAll( lReduced );
    }


    private BThreadVote getNextAllowedVote() {
	
	BThreadVote nextVote = lVotes.poll();
	List<BThreadVote> lReduced = new LinkedList<BThreadVote>();
	while ( blockedVotes.contains( nextVote )){
	    lReduced.add( nextVote );
	    nextVote = lVotes.poll();
	}
	blockedVotes.removeAll( lReduced );
	return nextVote;
    }


    @Override
    public void stop() {
	isActive = false;
	logger.info("Coordinator "+ getLevel() + " stoped.");
    }

    @Override
    public void vote(BThreadVote vote) {
	vote(vote, BThreadVoteType.ALLOW);
    }
    
    @Override
    public int getLevel() {
	return level;
    }
    
    public void executeAction(BThreadVote vote){
	BThreadType bthreadType = vote.getVoteType();
	
	BThreadManager.getInstance().getBThread( bthreadType ).setWeight( vote.getParameter() );
    }


    @Override
    public ITankInfo getTankInfo() {
	return new TankInfoImpl( medusa );
    }


    @Override
    public void vote(BThreadVote vote, BThreadVoteType voteType) {
	if (vote == null || vote.getVoteType().equals( BThreadType.DO_NOTHING )){ return; }
	
	try{
	    synchronizationLock.lock();
	    switch( voteType){
	    case ALLOW:
		lVotes.add( vote );
		break;
	    case BLOCK:
		blockedVotes.add( vote );
		break;
	    }
	    
	    synchronizationLock.unlock();
	}catch (Throwable t){}
	
    }

}
